//
//  asio_connection_handler.cpp
//  Amqp_test
//
//  Created by HNXF on 2022/7/26.
//


/**
 https://blog.csdn.net/weixin_42325783/article/details/123853093
 https://wenku.baidu.com/view/59c8726dcb50ad02de80d4d8d15abe23482f03c1.html
 */

#include "asio_connection_handler.h"
#include "amqp_buffer.h"
#include <iostream>
//#include <boost/bind.hpp>

asio_connection_handler::asio_connection_handler(
                        boost::asio::io_service& io_service,
                        const std::string& host,
                        uint16_t port)
    : io_service_(io_service)
    , socket_(io_service)
    , timer_(io_service)
    , input_buffer_(input_buffer_size_, 0)
    , amqp_buffer_(std::make_shared<amqp_buffer>(input_buffer_size_ * 2))
    , connection_ptr_(nullptr)
    , is_connected_(false)
    , is_writing_(false)
    , should_quit_(false)
{
    do_connect(host, port);
}

asio_connection_handler::~asio_connection_handler() {
    
}

void asio_connection_handler::connect_handler(const boost::system::error_code & ec)
{
    //如果ec返回成功我们就知道连接成功了
    //        if (!ec) {
    //            is_connected_ = true;
    //            do_read();
    //
    //            if (!output_buffer_.empty()) {
    //                do_write();
    //            }
    //        } else {
    //            on_network_error(ec, "connect");
    //        }
}

void asio_connection_handler::do_connect(const std::string& host, uint16_t port){
    using boost::asio::ip::tcp;

    // 抽象的是getaddrinfo()需要的参数
    tcp::resolver::query query(host, std::to_string(port));
    // 抽象的是getaddrinfo()的结果
    tcp::resolver::iterator iter = tcp::resolver(io_service_).resolve(query);
    // 程序会对定时器执行异步等待操作：
    timer_.expires_from_now(boost::posix_time::milliseconds(500));
    timer_.async_wait([this](const boost::system::error_code &ec){
        if (!ec && !is_connected_) {
            boost::system::error_code ignored_error;
            socket_.cancel(ignored_error);
            on_network_error(ec, "Connection timed out");
        }
    });

//    boost::asio::async_connect(socket_, iter, boost::bind(&asio_connection_handler::connect_handler, this, boost::asio::placeholders::error) );
    
    boost::asio::async_connect(socket_, iter,
        [this](boost::system::error_code ec, tcp::resolver::iterator)
    {
        if (!ec)
        {
            is_connected_ = true;
            do_read();

            if (!output_buffer_.empty())
            {
                do_write();
            }
        }
        else
        {
            on_network_error(ec, "connect");
        }
    });
    
}

void asio_connection_handler::do_read()
{
    std::cout << std::this_thread::get_id() << ": asio_connection_handler::do_read begin\n";
    socket_.async_receive(boost::asio::buffer(input_buffer_),
        [this](boost::system::error_code ec, std::size_t length)
    {
        std::cout << std::this_thread::get_id() << ": asio_connection_handler::do_read async_receive\n";
        if (!ec)
        {
            amqp_buffer_->write(input_buffer_.data(), length);
            handle_data();
            do_read();
        }
        else
        {
            on_network_error(ec, "read");
        }
    });
}
void asio_connection_handler::handle_data() {
    if (connection_ptr_ == nullptr) {
        return ;
    }
    const auto count = connection_ptr_->parse(amqp_buffer_->data(), amqp_buffer_->available());
    if (count == amqp_buffer_->available()) {
        amqp_buffer_->drain();
    } else if(count > 0) {
        amqp_buffer_->shl(count);
    }
}

void asio_connection_handler::do_write() {
    std::cout << std::this_thread::get_id() << ": asio_connection_handler::do_write\n";
    is_writing_ = true;
    boost::asio::async_write(socket_,
        boost::asio::buffer(output_buffer_.front()),
        [this](boost::system::error_code ec, std::size_t length) {
        if (!ec) {
            output_buffer_.pop_front();
            if (!output_buffer_.empty()) {
                do_write();
            } else {
                is_writing_ = false;
            }
            if (should_quit_) {
                socket_.close();
            }
        } else {
            boost::system::error_code ignored_error;
            socket_.close(ignored_error);
            on_network_error(ec, "write");
        }
    });
}

void asio_connection_handler::onData(
    AMQP::Connection* connection, const char* data, size_t size)
{
    connection_ptr_ = connection;

    output_buffer_.push_back(std::vector<char>(data, data + size));
    if (!is_writing_ && is_connected_)
    {
        do_write();
    }
}

void asio_connection_handler::onError(AMQP::Connection* connection, const char* message)
{
    std::cout << std::this_thread::get_id() << ": asio_connection_handler::onError: " << message << "\n";
    throw std::runtime_error(message);
}

void asio_connection_handler::onClosed(AMQP::Connection* connection)
{
    std::cout << std::this_thread::get_id() << ": asio_connection_handler::onClosed\n";
    should_quit_ = true;
    if (!is_writing_)
    {
        socket_.close();
    }
}

void asio_connection_handler::on_network_error(boost::system::error_code error_code, const std::string& source)
{
    std::cout << std::this_thread::get_id() << ": asio_connection_handler::on_network_error: " << error_code.message() << "(Source: " << source << ")\n";
    boost::asio::detail::throw_error(error_code);
}



